/*
 * Copyright (C) 2015 kmy-framework-base Project
 *               Author: Administrator
 *               Date: 2015年11月4日
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zmq.ha.client.resource;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.Callable;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AMQP.Queue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ShutdownSignalException;
import com.zmq.ha.client.core.Context;
import com.zmq.ha.client.declaration.Binding;
import com.zmq.ha.client.declaration.QueueDeclaration;
import com.zmq.ha.client.declaration.ResourceDeclaration;
import com.zmq.ha.client.exceptions.Exceptions;


/**
 * 
 * Name: RetryableSource.java
 * ProjectName: [kmy-framework-base]
 * Package: [com.kmy.framework.rabbitmq.recover.core.RetryableSource.java]
 * Description: rabbitMq恢复操作基类  
 * 
 * @since JDK1.7
 * @see
 *
 * Author: @author: Chris
 * Date: 2015年11月4日 下午3:43:43
 *
 * Update-User: @author
 * Update-Time:
 * Update-Remark:
 * 
 * Check-User:
 * Check-Time:
 * Check-Remark:
 * 
 * Company: kmy
 * Copyright: kmy
 */

public abstract class RetryableSource extends RetryableContext {
	
	public final Logger log = LoggerFactory.getLogger(getClass());

	/**
	 * @throws Exception 
	 * 
	 * @Title: deal
	 * @Description: 统一带结果的执行方法
	 * @param @param callable
	 * @param @return     
	 * @return T     
	 * @throws
	 */
	protected <T> T deal(Callable<T> callable, boolean isRetryMethod) throws Exception {
		while (true) {
			try {
				return callable.call();
			} catch (Exception e) {
				log.info("RMQ : ShutDownException occurs with RMQ , isRetryMethod : "+isRetryMethod);
				log.error("RMQ : RabbitMq Deal error with is RetryMethod:"+isRetryMethod, e );
				ShutdownSignalException se =  Exceptions.extractCause(e, ShutdownSignalException.class);
				//启动时候，RMQ-server没有启动,抛出错误 || 或者不可恢复的Exception
				if( se != null && (( !isRetryMethod) || (!Exceptions.isRetryable(se))) )
					throw e;
			    
				increaseRetryCount();
				if( isOverTetryCount() ){
					waitForRetry();
					continue;
				}
				else{
					//等待1.5s后继续执行
					Thread.sleep(Context.retryTime);
					continue;
				}
				
			}
		}
	}

	/** 获取当前需要恢复的Channel  */
	public abstract Channel getRecoveryChannel() throws IOException;

	/** 抛出恢复错误的异常  */
	public abstract boolean throwOnRecoveryFailure();

	/**
	 * 
	 * @Title: recoverExchange
	 * @Description: 恢复Exchange
	 * @param @param exchangeName
	 * @param @param exchangeDeclaration
	 * @param @throws Exception     
	 * @return void     
	 * @throws
	 */
	public void recoverExchange(String exchangeName, ResourceDeclaration exchangeDeclaration) throws Exception {
		try {
			exchangeDeclaration.invoke(getRecoveryChannel());
		} catch (Exception e) {
			if (throwOnRecoveryFailure() || Exceptions.isCausedByConnectionClosure(e))
				throw e;
		}
	}

	/**
	 * 
	 * @Title: recoverExchangeBindings
	 * @Description: 恢复Exchange的绑定
	 * @param @param exchangeBindings
	 * @param @throws Exception     
	 * @return void     
	 * @throws
	 */
	public void recoverExchangeBindings(Iterator<ArrayList<Binding>> exchangeBindings) throws Exception {
		if (exchangeBindings != null)
			synchronized (exchangeBindings) {
				while(exchangeBindings.hasNext()){
					ArrayList<Binding> bindings = exchangeBindings.next();
					for ( Binding binding : bindings)
						try {
							getRecoveryChannel().exchangeBind(binding.destination, binding.source, binding.routingKey,
									binding.arguments);
						} catch (Exception e) {
							if (throwOnRecoveryFailure() || Exceptions.isCausedByConnectionClosure(e))
								throw e;
						}
				}
			}
	}

	/**
	 * 
	 * @Title: recoverQueue
	 * @Description: 恢复queue
	 * @param @param queueName
	 * @param @param queueDeclaration
	 * @param @return
	 * @param @throws Exception     
	 * @return String     
	 * @throws
	 */
	public String recoverQueue(String queueName, QueueDeclaration queueDeclaration) throws Exception {
		try {
			String newQueueName = ((Queue.DeclareOk) queueDeclaration.invoke(getRecoveryChannel())).getQueue();
			if (queueName.equals(newQueueName)) {
			} else {
				queueDeclaration.queueName = newQueueName;
			}

			return newQueueName;
		} catch (Exception e) {
			if (throwOnRecoveryFailure() || Exceptions.isCausedByConnectionClosure(e))
				throw e;
			return queueName;
		}
	}

	/**
	 * 
	 * @Title: recoverQueueBindings
	 * @Description: 恢复queue的绑定
	 * @param @param queueBindings
	 * @param @throws Exception     
	 * @return void     
	 * @throws
	 */
	public void recoverQueueBindings(Iterator<ArrayList<Binding>> queueBindings) throws Exception {
		if (queueBindings != null)
			synchronized (queueBindings) {
				
				while(queueBindings.hasNext()){
					ArrayList<Binding> bindings = queueBindings.next();
				for (Binding binding : bindings)
					try {
							getRecoveryChannel().queueBind(binding.destination, binding.source, binding.routingKey,
									binding.arguments);
					} catch (Exception e) {
						if (throwOnRecoveryFailure() || Exceptions.isCausedByConnectionClosure(e))
							throw e;
					}
				}
			}
	}
}
